package com.asiainfo.iam.consumer;

import com.alibaba.fastjson.JSON;
import com.asiainfo.iam.constant.MQConstant;
import com.asiainfo.iam.entity.po.ModoTeam;
import com.asiainfo.iam.entity.po.ModoTeamMember;
import com.asiainfo.iam.entity.po.ModoTeamMemberRole;
import com.asiainfo.iam.entity.po.ModoUser;
import com.asiainfo.iam.entity.vo.MessageBase;
import com.asiainfo.iam.entity.vo.Organization;
import com.asiainfo.iam.entity.vo.Role;
import com.asiainfo.iam.entity.vo.User;
import com.asiainfo.iam.mapper.ModoTeamMapper;
import com.asiainfo.iam.mapper.ModoTeamMemberMapper;
import com.asiainfo.iam.mapper.ModoTeamMemberRoleMapper;
import com.asiainfo.iam.mapper.ModoUserMapper;
import com.asiainfo.iam.tool.BloomFilterUtil;
import com.asiainfo.iam.tool.MsgDigesUtil;
import com.asiainfo.iam.tool.RedisMsgIdUtil;
import com.asiainfo.iam.tool.ToPo;
import com.fasterxml.jackson.core.JsonParseException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.dao.DuplicateKeyException;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 采用集群消费模式（默认）
 * 消费者从 Broker 拉取消息
 * 处理消息（如写入数据库）
 * 向 Broker 发送消费确认（ConsumeOrderlyStatus.SUCCESS）
 * 关键逻辑：
 * 若确认成功，消息从队列中移除
 * 若未确认（如消费异常），消息会重新入队，供其他消费者或当前消费者重试
 */
@Configuration
// TODO 3
@ConditionalOnProperty(name = "enable.syn.controller", havingValue = "true")
public class DataSyncConsumer {
    private static final Logger log = LoggerFactory.getLogger(DataSyncConsumer.class);

    private DefaultMQPushConsumer consumer;

    @Value("${rocketmq.namesrv.address}")
    private String namesrvAddr;
    // 使用统一的消费者组
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Autowired
    private ModoTeamMapper modoTeamMapper;
    @Autowired
    private ModoUserMapper modoUserMapper;
    @Autowired
    ModoTeamMemberRoleMapper modoTeamMemberRoleMapper;
    @Autowired
    private ModoTeamMemberMapper modoTeamMemberMapper;

    @Autowired
    private RedisMsgIdUtil redisMsgIdUtil;

    @Autowired
    private MsgDigesUtil msgDigesUtil;

    @Autowired
    private BloomFilterUtil bloomFilterUtil;

    @Autowired
    private ToPo toPo;

//    @Bean(name = "MQPushConsumer")
    @Bean(destroyMethod ="shutdown")
    public DefaultMQPushConsumer defaultMQPushConsumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        // 从上次消费位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 同时订阅全量和增量数据的Tag
        consumer.subscribe(MQConstant.DATA_SYNC_TOPIC,
                MQConstant.FULL_SYNC_TAG + "||" + MQConstant.INCREMENTAL_SYNC_TAG);

        //顺序消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                // TODO 增加消费者重试次数、增强消息幂等性，避免重复处理
                // 关闭自动提交（顺序消费默认自动提交，此处显式控制）
                context.setAutoCommit(false);
                // 重试最大次数（可配置在配置文件中）
                int maxReconsumeTimes = 3;

                    for (MessageExt msg : msgs) {
                        String msgId = msg.getMsgId(); // 消息唯一ID（用于幂等,可redis实现）
                        int reconsumeTimes = msg.getReconsumeTimes(); // 当前重试次数
                        log.info("消息[{}]第{}次消费",msgId, reconsumeTimes);
                        try {


                            // 处理消息
                            String msgBody = new String(msg.getBody(), StandardCharsets.UTF_8);// 消息内容
                            // 计算消息体hash值
                            String contentHash = msgDigesUtil.calculateHash(msgBody);
                            // 通过布隆过滤器判断消息是否已经处理过
                            if(bloomFilterUtil.mightContain(contentHash)){
                                log.info("消息重复或已经处理过，布隆过滤成功！");
                                continue;
                            }
//                            // 查询redis判断消息是否已经处理过（不能应对高并发）
//                            if (redisMsgIdUtil.isMessageProcessed(msgId)){
//                                log.info("消息[{}]已经处理过，跳过",msgId);
//                                continue;
//                            }
                            // 应对高并发
                            if (!redisMsgIdUtil.checkAndMarkMessageProcessed(contentHash)) {
                                log.info("相同消息已经处理过，跳过");
                                continue;
                            }
                            String tags = msg.getTags();// 消息标签
                            //解析消息
                            MessageBase<?> baseMsg = JSON.parseObject(msgBody, MessageBase.class);
                            // 获取数据类型
                            String objectType = baseMsg.getMsgObject();
                            // 获取操作类型
                            String operateType = baseMsg.getMsgOperateType();

                            // 根据Tag区分全量和增量数据
                            if (MQConstant.FULL_SYNC_TAG.equals(tags)) {
                                // 全量数据处理逻辑
                                handleFullData(objectType, baseMsg);
                            } else if (MQConstant.INCREMENTAL_SYNC_TAG.equals(tags)) {
                                // 增量数据处理逻辑，需要操作类型
                                handleIncrementalData(objectType, operateType, baseMsg);
                            } else {
                                log.warn("未知标签类型: {}", tags);
                            }
                            //处理成功，提交偏移量
                            context.setAutoCommit(true);
//                            // TODO 添加redis记录已经消费的消息
//                            redisMsgIdUtil.markMessageProcessed(msgId);
//                            log.info("msgId已经记录到redis！");
                            // TODO 添加布隆过滤器
                            bloomFilterUtil.put(contentHash);
                             log.info("消息体已经记录到布隆过滤器！");
                            //向 Broker 发送消费确认,若确认成功，消息从队列中移除
                            return ConsumeOrderlyStatus.SUCCESS;

                        }catch (Exception e){
                            log.error("消息[{}]消费失败（第{}次）",msgId, reconsumeTimes,e);

                            //判断是否超过最大重试次数
                            if(reconsumeTimes>=maxReconsumeTimes){
                                //超过最大重试次数，记录死信日志（或发送到死信队列）
                                log.error("消息[{}]已超过最大重试次数({}次)，已记录为死信",msgId,maxReconsumeTimes);
                                // TODO 将消息发送到死信队列
                                // sendToDlq(msg);
                                // 手动提交偏移量，避免重复消费
                                context.setAutoCommit(true);
                                return ConsumeOrderlyStatus.SUCCESS;
                            }
                            // 区分异常类型：可重试异常继续重试，不可重试异常直接标记成功（避免无限重试）
                            if (isUnrecoverableException(e)){
                                log.error("消息[{}]发生不可重试异常，终止重试", msgId);
                                context.setAutoCommit(true);
                                return ConsumeOrderlyStatus.SUCCESS;
                            }else {
                                // 可重试异常，暂停队列后重试（延长重试间隔，默认1秒，可调整）
                                context.setSuspendCurrentQueueTimeMillis(3000); //3秒后重试
                                return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;//暂停队列
                            }

                        }
                    }
                return ConsumeOrderlyStatus.SUCCESS;
                }
                //判断是否为不可重试异常（根据业务场景扩展）
            private boolean isUnrecoverableException(Exception e) {
                // 判断异常类型，是否为不可重试异常
                // 例如，判断是否为主键冲突、数据异常等
                return e instanceof DuplicateKeyException
                        || e instanceof JsonParseException
                        || e.getMessage().contains("数据格式错误")
                        || e.getMessage().contains("唯一性约束条件");
            }
        });

        consumer.start();
        log.info("数据同步消费者启动成功！");
        return consumer;

    }
    /**
     * 处理全量数据
     */
    private void handleFullData(String objectType, MessageBase<?> baseMsg) {
        log.info("处理全量数据: {}", objectType);
        Map<String, String> memberData = new HashMap<>();
        switch (objectType) {
            case MQConstant.OBJECT_ORG:
                //将JSON数据转为PO
                Organization org = JSON.parseObject(
                        JSON.toJSONString(baseMsg.getMsgData()),
                        Organization.class
                );
                handleOrgAdd(org);
                break;
            case MQConstant.OBJECT_ROLE:
                Role role = JSON.parseObject(
                        JSON.toJSONString(baseMsg.getMsgData()),
                        Role.class
                );
                handleRoleAdd(role);
                memberData.put("roleName", role.getRoleName());
                break;

            case MQConstant.OBJECT_USER:
                User user = JSON.parseObject(
                        JSON.toJSONString(baseMsg.getMsgData()),
                        User.class
                );
                handleUserAdd(user);
                memberData.put("userId", user.getAccount());
                memberData.put("teamName", user.getOrgName());
                break;
            default:
                log.warn("未知对象类型: {}", objectType);
        }
        //更新三者之间关系表
        handleMemberAdd(memberData.get("teamName"),
                memberData.get("userId"),
                memberData.get("roleName"));
    }
    /**
     * 处理增量数据
     */
    private void handleIncrementalData(String objectType, String operateType, MessageBase<?> baseMsg) {
        log.info("处理增量数据: {}, 操作类型: {}", objectType, operateType);
        // 1. 新增操作：按 组织→角色→用户 顺序处理（确保依赖对象先存在）
        if (MQConstant.OPERATE_ADD.equals(operateType)) {
            //存储三者
            Map<String, String> memberData = new HashMap<>();
            // 强制按组织→角色→用户的顺序处理，即使消息顺序错乱也按此规则执行
            switch (objectType) {
                case MQConstant.OBJECT_ORG:
                    Organization org = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            Organization.class
                    );
                    handleOrgAdd(org);  // 先处理组织新增
                    break;
                case MQConstant.OBJECT_ROLE:
                    Role role = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            Role.class
                    );
                    handleRoleAdd(role);  // 再处理角色新增
                    memberData.put("roleName", role.getRoleName());
                    break;
                case MQConstant.OBJECT_USER:
                    User user = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            User.class
                    );
                    handleUserAdd(user);  // 最后处理用户新增
                    memberData.put("userId", user.getAccount());
                    memberData.put("teamName", user.getOrgName());
                    break;
                default:
                    log.warn("未知新增对象类型: {}", objectType);
            }
            //更新三者之间关系表
            handleMemberAdd(memberData.get("teamName"),
                    memberData.get("userId"),
                    memberData.get("roleName"));

        }
        //2.处理更新操作
        else if (MQConstant.OPERATE_UPDATE.equals(operateType)) {
            //存储三者
            Map<String, String> memberDataUpdate = new HashMap<>();
            switch (objectType) {
                case MQConstant.OBJECT_USER:
                    User user = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            User.class
                    );
                    handleUserUpdate(user);
                    memberDataUpdate.put("userId", user.getAccount());
                    memberDataUpdate.put("teamName", user.getOrgName());
                    break;
                case MQConstant.OBJECT_ROLE:
                    Role role = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            Role.class
                    );
                    handleRoleUpdate(role);
                    memberDataUpdate.put("roleName", role.getRoleName());
                    break;
                case MQConstant.OBJECT_ORG:
                    Organization org = JSON.parseObject(
                            JSON.toJSONString(baseMsg.getMsgData()),
                            Organization.class
                    );
                    handleOrgUpdate(org);
                    break;
                default:
                    log.warn("未知修改对象类型: {}", objectType);
            }
            handleMemberUpdate(memberDataUpdate.get("teamName"),
                    memberDataUpdate.get("userId"),
                    memberDataUpdate.get("roleName"));

        }
        // 3. 处理删除操作：按 用户→角色→组织 顺序处理（避免依赖冲突）
        else if ( MQConstant.OPERATE_DELETE.equals(operateType)) {
            // 强制按用户→角色→组织的顺序处理
            switch (objectType) {
                case MQConstant.OBJECT_USER:
                    handleUserDelete(baseMsg);
                    break;
                case MQConstant.OBJECT_ROLE:
                    handleRoleDelete(baseMsg);
                    break;
                case MQConstant.OBJECT_ORG:
                    handleOrgDelete(baseMsg);  // 组织删除需特殊处理（先删用户）
                    break;
                default:
                    log.warn("未知修改/删除对象类型: {}", objectType);
            }
        } else {
            log.warn("不支持的操作类型: {}", operateType);
        }
    }

    /**
     * 新增团队 - 全量/增量
     */
    private void handleOrgAdd(Organization org) {
        // 全量模式下通常是覆盖更新
        log.info("同步新增组织: {}", org.getId());
        try {
            ModoTeam existing = modoTeamMapper.findById(org.getId());
            ModoTeam modoTeam = toPo.toModoTeam(org);
            if(existing == null){
                modoTeamMapper.insert(modoTeam);
                log.info("团队新增成功!");
            }else {
                log.info("团队[{}]已存在！", org.getId());
            }
        } catch (DuplicateKeyException e) {
            // 记录错误并抛出，由外层统一处理
            log.warn("唯一键冲突，用户已存在！");
        }catch (Exception e) {
            throw new RuntimeException("其他错误", e);
        }

    }
    /**
     * 新增用户 - 全量/增量
     */
    private void handleUserAdd(User user) {
        log.info("同步新增用户: {}", user.getAccount());
        try {
            ModoUser existing = modoUserMapper.findById(user.getAccount());
            ModoUser modoUser = toPo.toModoUser(user);
            if(existing == null){
                modoUserMapper.insert(modoUser);
                log.info("用户新增成功!");
            }else {
                log.info("用户[{}]已存在！", user.getAccount());
            }
        } catch (DuplicateKeyException e) {
            // 捕获唯一键异常，忽略
            log.warn("唯一键冲突，用户已存在！");
        }catch (Exception e) {
            throw new RuntimeException("其他错误", e);
        }
    }

    /**
     * 新增角色 - 全量/增量
     */
    private void handleRoleAdd(Role role) {
        // 全量模式下通常是覆盖更新
        log.info("同步新增角色: {}", role.getRoleName());
        try {
            ModoTeamMemberRole existing = modoTeamMemberRoleMapper.findByName(role.getRoleName());
            ModoTeamMemberRole modoTeamMemberRole = toPo.toModoTeamMemberRole(role);

            if (existing == null) {
                modoTeamMemberRoleMapper.insert(modoTeamMemberRole);
                log.info("角色新增成功！ ");
            }else {
                log.info("角色[{}]已存在！", role.getRoleName());
            }
        } catch (DuplicateKeyException e) {
            // 捕获唯一键异常，忽略
            log.warn("唯一键冲突，角色已存在！");
//            throw new RuntimeException("角色同步失败", e);
        }catch (Exception e) {
            throw new RuntimeException("其他错误", e);
        }
    }

    /**
     * 新增三者关系表 - 全量/增量
     */

    private void handleMemberAdd(String teamName,String userId,String roleName) {
        log.info("新增关系表");
        try {
            if (userId!=null){
                ModoTeamMember existing = modoTeamMemberMapper.findById(userId);
                ModoTeamMember modoTeamMember = toPo.toModoTeamMember(teamName, userId, roleName);
                if (existing == null) {
                    modoTeamMemberMapper.insert(modoTeamMember);
                    log.info("关系表新增成功！");
                }else {
                    log.info("关系表已存在！");
                }
            }else {
                log.info("没有用户需要加入团队");
            }

        } catch (DuplicateKeyException e) {
            // 捕获唯一键异常，忽略
            log.warn("唯一键冲突，关系已存在");
        }catch (Exception e) {
            throw new RuntimeException("其他错误", e);
        }

    }

    /**
     * 增量--更新用户
     */
    private void handleUserUpdate(User user) {
        log.info("开始更新用户: {}", user.getAccount());
        try {
            ModoUser existingUser = modoUserMapper.findById(user.getAccount());
            ModoUser modoUser = toPo.toModoUser(user);
            if (existingUser != null) {
                modoUserMapper.update(modoUser);
                log.info("用户更新成功！ ");
            }else {
                log.info("用户不存在！ ");
            }
        }
        catch (Exception e) {
            log.error("用户更新失败: {}", e);
            throw new RuntimeException("用户更新失败", e);
        }
    }

    /**
     * 增量--更新角色
     */
    private void handleRoleUpdate(Role role) {
        log.info("开始更新角色: {}", role.getRoleName());
        try {
            ModoTeamMemberRole existingRole = modoTeamMemberRoleMapper.findByName(role.getRoleName());
            ModoTeamMemberRole modoTeamMemberRole = toPo.toModoTeamMemberRole(role);
            if (existingRole != null) {
                modoTeamMemberRoleMapper.update(modoTeamMemberRole);
                log.info("角色更新成功！ ");
            }else {
                log.info("角色不存在！ ");
            }
        } catch (Exception e) {
            log.error("角色更新失败: {}", e);
            throw new RuntimeException("角色更新失败", e);
        }
    }

    /**
     * 增量--更新组织
     */
    private void handleOrgUpdate(Organization org) {
        log.info("开始更新团队: {}", org.getId());
        try {
            ModoTeam existing = modoTeamMapper.findById(org.getId());
            ModoTeam modoTeam = toPo.toModoTeam(org);
            if (existing != null) {
                modoTeamMapper.update(modoTeam);
                log.info("团队更新成功！ ");
            }else {
                log.info("团队不存在！ ");
            }
        }
        catch (Exception e) {
            log.error("团队更新失败: {}", e);
            throw new RuntimeException("团队更新失败", e);
        }
    }

    /**
     *增量--更新关系表
     */
    private void handleMemberUpdate(String teamName,String userId,String roleName) {
        log.info("开始更新关系表");
        try {
            if(userId!=null){
                ModoTeamMember existing = modoTeamMemberMapper.findById(userId);
                ModoTeamMember modoTeamMember = toPo.toModoTeamMember(teamName, userId, roleName);
                if (existing != null) {
                    modoTeamMemberMapper.update(modoTeamMember);
                    log.info("关系表更新成功！ ");
                }else {
                    log.info("关系表不存在！ ");
                }
            }else{
                log.info("没有用户需要更新");
            }

        }
        catch (Exception e) {
            log.error("关系表更新失败: {}", e);
            throw new RuntimeException("关系表更新失败", e);
        }
    }

    /**
     * 增量--删除用户
     */
    private void handleUserDelete(MessageBase<?> message) {
        User user = JSON.parseObject(
                JSON.toJSONString(message.getMsgData()),
                User.class
        );
        log.info("开始删除用户: {}", user.getId());
        try {
            ModoUser existingUser = modoUserMapper.findById(user.getAccount());
            ModoTeamMember existingMember = modoTeamMemberMapper.findById(user.getAccount());
            if (existingUser != null) {
                modoUserMapper.deleteById(user.getAccount());
                log.info("用户删除成功！");
                if(existingMember != null){
                    modoTeamMemberMapper.deleteById(user.getAccount());
                    log.info("移除用户表关系！");
                }else {
                    log.info("该用户未加入团队！");
                }
            }else {
                log.info("用户不存在！");
            }
        } catch (Exception e) {
            log.error("用户删除失败: {}", e);
            throw new RuntimeException("用户删除失败", e);
        }
    }
    /**
     * 增量--删除角色
     */
    private void handleRoleDelete(MessageBase<?> message) {
        Role role = JSON.parseObject(
                JSON.toJSONString(message.getMsgData()),
                Role.class
        );
        log.info("开始删除角色: {}", role.getRoleName());
        try {
            ModoTeamMemberRole existing = modoTeamMemberRoleMapper.findByName(role.getRoleName());
            if (existing != null) {
                modoTeamMemberRoleMapper.deleteByName(role.getRoleName());
                log.info("角色删除成功！");
            }else{
                log.info("角色不存在！");
            }
        }
        catch (Exception e) {
            log.error("角色删除失败: {}", e);
            throw new RuntimeException("角色删除失败", e);
        }
    }

    /**
     * 增量--删除组织
     * 处理组织删除：先检查该组织下是否有用户，有则先删用户，再删组织
     */
    private void handleOrgDelete(MessageBase<?> message) {
        Organization org = JSON.parseObject(
                JSON.toJSONString(message.getMsgData()),
                Organization.class
        );
        String teamName = org.getOrgName();
        String orgId = org.getId();
        log.info("开始处理团队删除，团队名字: {}", teamName);

        try {
            // 步骤1：查询该组织下的所有用户
            List<ModoTeamMember> allUser = modoTeamMemberMapper.findAllUser(teamName);  // 需在ModoUserMapper中新增此方法
            log.info("查询到团队[{}]下有{}个用户", teamName, allUser.size());

            // 步骤2：若有用户，先删除这些用户
            if (!allUser.isEmpty()) {
                for (ModoTeamMember member : allUser) {
                    //1.先删用户表的用户
                    modoUserMapper.deleteById(member.getUserId());  // 删除单个用户
                    //2.再删关系表的绑定关系
                    modoTeamMemberMapper.deleteById(member.getUserId());
                    log.info("已删除团队[{}]下的用户，用户ID: {}", teamName, member.getUserId());
                }
            }else {
                log.info("团队[{}]下无用户，直接删除团队", teamName);
            }

            // 步骤3：删除组织
            ModoTeam existingOrg = modoTeamMapper.findById(orgId);
            if (existingOrg != null) {
                modoTeamMapper.deleteById(orgId);
                log.info("团队[{}]删除成功", orgId);
            } else {
                log.info("团队[{}]不存在，无需删除", orgId);
            }

        } catch (Exception e) {
            log.error("团队[{}]删除失败", orgId, e);
            throw new RuntimeException("团队删除失败（含用户联动处理）", e);  // 事务回滚
        }
    }
    @PostConstruct
    public void init() {
        log.info("消费者初始化完成，等待接收数据...");
    }

    @PreDestroy
    public void shutdown() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("消费者已优雅关闭");
        }
    }
}

